package com.zhai.cloud.biz.deal.consumer;

import com.rabbitmq.client.Channel;


import com.zhai.cloud.biz.deal.domain.BizDealSave;
import com.zhai.cloud.common.util.DateUtil;
import com.zhai.cloud.common.util.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;

import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Component;



/**
 * @author Created by 翟凤玺
 * @package com.zhai.cloud.deal.consumer
 * @project-name zhai-cloud
 * @description 消费者-交易保存-业务
 * @create-datetime 2020-5-6 11:15
 */
@Slf4j
@Component
public class BizDealSaveConsumer {

    @RabbitListener(
            bindings = @QueueBinding(
                value = @Queue(
                        value = "${biz.mq.deal-save.queue.name}",
                        durable = "${biz.mq.deal-save.queue.durable}",
                        autoDelete = "biz.mq.deal-save.queue.auto-delete"),
                exchange = @Exchange(
                        value = "${biz.mq.deal-save.exchange.name}",
                        durable = "${biz.mq.deal-save.exchange.durable}",
                        type = "${biz.mq.deal-save.exchange.type}",
                        ignoreDeclarationExceptions = "${biz.mq.deal-save.exchange.ignoreDeclarationExceptions}"),
                key = "${biz.mq.deal-save.key}"
            )
    )
    @RabbitHandler
    public void onMessage(Message message, Channel channel) {
        log.info("消费者入参 message【消息】:{}, channel【】:{}", JSONUtil.toStr(message), JSONUtil.toStr(channel));
        MessageHeaders headers = message.getHeaders();
        Object body = message.getPayload();
        BizDealSave bizDealSave = JSONUtil.toObj(body.toString(), BizDealSave.class);
        log.info(DateUtil.fmtLocalDateTimeToStr(DateUtil.getCurrentLocalDateTime(), DateUtil.DATETIME_FMT_3) + "消费机接收时间" + "  流水号：" + bizDealSave.getSerialNum());
        // 业务处理
        // 手动ACK
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        try {
            log.info(DateUtil.fmtLocalDateTimeToStr(DateUtil.getCurrentLocalDateTime(), DateUtil.DATETIME_FMT_3) + "消费机业务处理开始时间"+ "  流水号：" + bizDealSave.getSerialNum());
            Thread.sleep(5000);
            log.info(DateUtil.fmtLocalDateTimeToStr(DateUtil.getCurrentLocalDateTime(), DateUtil.DATETIME_FMT_3) +"业务处理完成"+ "  流水号：" + bizDealSave.getSerialNum());
            // 手动签收-成功确认
            // deliveryTag： 该消息的index
            // multiple：是否批量. true：将一次性ack所有小于deliveryTag的消息。
            channel.basicAck(deliveryTag, false);
//            throw new Exception("报错啦");
        } catch (Exception e) {
            try {
                // 手动签收-失败确认
                // deliveryTag： 该消息的index
                // multiple：是否批量. true：将一次性拒绝所有小于deliveryTag的消息。
                // requeue：被拒绝的是否重新入队列。
                // 提示：channel.basicNack 与 channel.basicReject 的区别在于basicNack可以批量拒绝多条消息，而basicReject一次只能拒绝一条消息。
                channel.basicNack(deliveryTag, false, false);
            } catch (Exception e1) {
                e1.printStackTrace();
            }
        }
    }

    /**
     * 注意：消息手动拒绝中如果requeue为true会重新放入队列，但是如果消费者在处理过程中一直抛出异常，会导致入队-》拒绝-》入队的循环
     * 第一种方法：根据异常类型来选择是否重新放入队列。
     * 第二种方法：先成功确认，然后通过channel.basicPublish()重新发布这个消息。重新发布的消息网上说会放到队列后面，进而不会影响已经进入队列的消息处理。
     * void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body)
     */

}
